package com.example.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.kafka.data.BaseEventData;
import com.example.kafka.handler.EventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

@Service
public class KafkaConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerService.class);

    private final EventHandlerFactory eventHandlerFactory;

    @Autowired
    public KafkaConsumerService(EventHandlerFactory eventHandlerFactory) {
        this.eventHandlerFactory = eventHandlerFactory;
    }

    private static final String TRACE_ID_KEY = "CT-REQUESTKEY";

    @KafkaListener(topics = "#{@kafkaListenerInitializer.topics}", groupId = "${spring.kafka.consumer.group-id}")
    public void listenToMessages(String message) {
        String traceId = UUID.randomUUID().toString();
        try {
            MDC.put(TRACE_ID_KEY, traceId);
            handleSingleMessage(message);
        } catch (Exception e) {
            logger.error("Failed to process message: {}", message, e);
        } finally {
            MDC.remove(TRACE_ID_KEY);
        }
    }

    public void handleSingleMessage(String message) {
        try {
            logger.info("Received message: {}", message);
            JSONObject jsonObject = JSON.parseObject(message);
            Integer eventType = jsonObject.getInteger("eventType");

            EventHandler<? extends BaseEventData> handler = eventHandlerFactory.getHandlerForEventType(eventType);
            if (handler != null) {
                Class<? extends BaseEventData> dataType = handler.getEventDataType();

                KafkaEvent event = JSON.parseObject(message, KafkaEvent.class);
                String dataJson = jsonObject.getString("data");

                List<? extends BaseEventData> parsedDataList = JSON.parseArray(dataJson, dataType);
                List<BaseEventData> dataList = new ArrayList<>(parsedDataList);
                event.setData(dataList);
                eventHandlerFactory.processEvent(event);
                logger.info("Event processing completed for eventType: {}", eventType);
            } else {
                logger.warn("No handler found for event type: {}", eventType);
            }

        } catch (Exception e) {
            logger.error("Failed to process message: {}", message, e);
        }
    }
}